package com.xxx.rpc.registry.zookeeper;

import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.xxx.rpc.client.NettyRpcClient;
import com.xxx.rpc.client.RpcClient;
import com.xxx.rpc.common.bean.IpPort;
import com.xxx.rpc.common.util.StringUtil;
//import com.xxx.rpc.registry.RpcClient;
import com.xxx.rpc.registry.ServiceDiscovery;

/**
 * 基于 ZooKeeper 的服务发现接口实现
 *
 * @author huangyong & Rosun
 * @since 1.0.0
 */
public class ZooKeeperServiceDiscovery implements ServiceDiscovery {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZooKeeperServiceDiscovery.class);

    private String zkAddress;
    
    private ServiceCenter serviceCenter;

    public ZooKeeperServiceDiscovery(String zkAddress) {
        this.zkAddress = zkAddress;
        this.serviceCenter = new ZookeeperServiceCenter(this.zkAddress);
        try {
			this.serviceCenter.createWatchClient();
		} catch (Exception e) {
			e.printStackTrace();
			LOGGER.error("CAN NOT create watch client . please check zookeeper url :" + zkAddress);
		}
    }

    
    
    

	/**
     * 此方法有严重性能问题，已被罗珊弃用。
     * @see com.xxx.rpc.registry.ServiceDiscovery#discover(java.lang.String)
     */
//    @Override
//    public String discover(String name) {
//    	LOGGER.warn("此方法有严重性能问题，已被罗珊弃用。");
//        // 创建 ZooKeeper 客户端
//        ZkClient zkClient = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
//        LOGGER.debug("connect zookeeper");
//        try {
//            // 获取 service 节点
//            String servicePath = Constant.ZK_REGISTRY_PATH + "/" + name;
//            if (!zkClient.exists(servicePath)) {
//                throw new RuntimeException(String.format("can not find any service node on path: %s", servicePath));
//            }
//            List<String> addressList = zkClient.getChildren(servicePath);
//            if (CollectionUtil.isEmpty(addressList)) {
//                throw new RuntimeException(String.format("can not find any address node on path: %s", servicePath));
//            }
//            // 获取 address 节点
//            String address;
//            int size = addressList.size();
//            if (size == 1) {
//                // 若只有一个地址，则获取该地址
//                address = addressList.get(0);
//                LOGGER.debug("get only address node: {}", address);
//            } else {
//                // 若存在多个地址，则随机获取一个地址
//                address = addressList.get(ThreadLocalRandom.current().nextInt(size));
//                LOGGER.debug("get random address node: {}", address);
//            }
//            // 获取 address 节点的值
//            String addressPath = servicePath + "/" + address;
//            return zkClient.readData(addressPath);
//        } finally {
//            zkClient.close();
//        }
//    }

     
     
    
    
//    private ReadWriteLock nsMapLock = new ReentrantReadWriteLock();
    private Lock clientsLock = new ReentrantLock();
    
//    /**
//     * key:nameWithVersion   xxxservice-version1 <br>
//     * value:serviceAddress 集合；   [127.0.0.1:9876,127.0.0.2:9875]   一个接口 可能有多个节点来提供服务！
//     */
//    private MultiValueMap<String ,String> nameServiceMap  = MapUtils.multiValueMap(new ConcurrentHashMap<String,Collection<String>>());
    
    /**
     * key:     serviceAddress   127.0.0.1:9876<br>
     * value：     RpcClient  ；一个ip端口  对应一个RPC客户端（长链接）
     */
    private Map<IpPort,RpcClient> clientsMap = new ConcurrentHashMap<IpPort,RpcClient>();
    
    private long point = 0;
    
	@Override
	public RpcClient discover(String serviceName, String serviceVersion) throws Exception{
		RpcClient rpcClient = null;
		String nameWithVersion = serviceName;
        if (StringUtil.isNotEmpty(serviceVersion)) {
        	nameWithVersion += "-" + serviceVersion;
        }
        /****
        String serviceAddress = null;
        try{
        	nsMapLock.readLock().lock();
        	Collection<IpPort> coll = this.nameServiceMap.getCollection(nameWithVersion);
        	if(coll != null && coll.size() > 0){
        		int size = coll.size();
            	int random = ThreadLocalRandom.current().nextInt(size);
            	serviceAddress = (String) coll.toArray()[random];
        	}
        }finally{
        	nsMapLock.readLock().unlock();
        	if(serviceAddress == null){
        		serviceAddress = this.findServerIPandPort(nameWithVersion);
        		if(serviceAddress != null){
        			try{
        				nsMapLock.writeLock().lock();
        				if(!this.nameServiceMap.containsKey(nameWithVersion) 
        						|| !this.nameServiceMap.getCollection(nameWithVersion).contains(nameWithVersion)){
        					this.nameServiceMap.put(nameWithVersion, serviceAddress);
        				}
        			}finally{
        				nsMapLock.writeLock().unlock();
        			}
        		}
        	}
        }
        
        if (StringUtil.isEmpty(serviceAddress)) {
            throw new RuntimeException("server address is empty ,serviceName="+serviceName+",serviceVersion="+serviceVersion);
        }else{
        	LOGGER.debug("findServerIPandPort service: {} => {}", nameWithVersion, serviceAddress);
        }
        ****/
        List<IpPort> serverList = this.serviceCenter.getServerStub(nameWithVersion);
        if (serverList == null || serverList.size() == 0) {
            throw new RuntimeException("Server not ready,serviceName="+serviceName+",serviceVersion="+serviceVersion);
        }else{
        	LOGGER.debug("findServerIPandPort service: {} => {}", nameWithVersion, serverList.toArray().toString());
        }
        //负载均衡算法，轮询
        int size = serverList.size();
        long mode = (point++) % size;
        IpPort serviceAddress = serverList.get((int)mode);
        //负载均衡算法，轮询
        LOGGER.debug("简单轮询负载均衡 >>> " + serviceAddress.toString());
        //复用一个客户端  或者新建一个客户端
        try{
        	clientsLock.lock();
        	//罗珊  先尝试从clientsMap 中取一个相应的client
            if(clientsMap.containsKey(serviceAddress)){
            	//优先 复用一个客户端
            	rpcClient = clientsMap.get(serviceAddress);
            }else{
            	 // 从 RPC 服务地址中解析主机名与端口号
                // 创建 RPC 客户端对象并发送 RPC 请求
                rpcClient = new NettyRpcClient();
    			rpcClient.initLongConnection(serviceAddress.getIp(), serviceAddress.getPort());
    			rpcClient.setServiceDiscoveryPoint(this);
    			//链接成功后 装到MAP中；
    			clientsMap.put(serviceAddress, rpcClient);
            }
        }catch(Exception ex){
        	rpcClient = null;
        	throw ex;
        }finally{
        	clientsLock.unlock();
        }
		return rpcClient;
	}

	@Override
	public void rpcClientCloseCallBack(RpcClient closingRpcClient) {
		if(closingRpcClient == null){
			LOGGER.warn("closingRpcClient is null,ignore rpcClientCloseCallBack .");
			return;
		}
		try{
        	clientsLock.lock();
        	//罗珊  先尝试从clientsMap 中取一个相应的client
            if(clientsMap.values().contains(closingRpcClient)){
            	Set<Entry<IpPort,RpcClient>> set = clientsMap.entrySet();
            	for(Entry<IpPort,RpcClient> entry : set){
            		if(entry.getValue().equals(closingRpcClient)){
            			clientsMap.remove(entry.getKey());
            		}
            	}
            } 
        }catch(Exception ex){
        	LOGGER.error(ex.getMessage(),ex);
        }finally{
        	clientsLock.unlock();
        }
		LOGGER.info("#### remove closing rpc client ,end rpcClientCloseCallBack .");
	}
}